package com.study.sbsummary.config.kafka.consumer;

import com.study.sbsummary.config.kafka.Topics;
import com.study.sbsummary.utils.ThreadUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;

@Data
@Slf4j
public abstract class AbstractConsumerThread implements IConsumerThread {
    private Consumer consumer;
    private Topics.Topic topic;
    private IRecordsHandler handler;
    private int index;
    private Properties properties;

    public AbstractConsumerThread(Consumer consumer, Topics.Topic topic, IRecordsHandler handler, int index,
                                  Properties properties) {
        this.consumer = consumer;
        this.topic = topic;
        this.handler = handler;
        this.index = index;
        this.properties = properties;
    }

    @Override
    public Consumer consumer() {
        return consumer;
    }

    @Override
    public Topics.Topic topic() {
        return topic;
    }

    @Override
    public IRecordsHandler handler() {
        return handler;
    }

    @Override
    public void run() {
        Thread.currentThread().setName(topic.getName() + "-consumer-" + index);
        Duration duration = Duration.ofMillis(topic.getDuration());
        IRecordsHandler handler = handler();
        while (true) {
            ConsumerRecords consumerRecords = consumer.poll(duration);
            if (consumerRecords != null && !consumerRecords.isEmpty()) {
                Iterator iterator = consumerRecords.iterator();
                while (iterator.hasNext()) {
                    ConsumerRecord consumerRecord = (ConsumerRecord)iterator.next();
                    try {
                        for (int i = 1; i <= 3; i++) {
                            try {
                                if (!handler.handle(consumerRecord)) {
                                    log.error("failed to handle message {}", consumerRecord.value());
                                }
                                break;
                            } catch (Exception e) {
                                log.error("failed to handle message {} after {} times and will skip after 3 times, " +
                                                "due to {}", i, consumerRecord.value(), e);
                            }
                            ThreadUtil.sleep(1000);
                        }
                        if (!topic().isAutoCommit()) {
                            Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
                            TopicPartition topicPartition = new TopicPartition(consumerRecord.topic(), consumerRecord.partition());
                            offsets.put(topicPartition, new OffsetAndMetadata(consumerRecord.offset()));
                            consumer.commitSync(offsets);
                        }
                    } catch (Exception e) {
                        log.error("{}",e.getMessage());
                    }
                }
            } else {
                ThreadUtil.sleep(3 * 1000);
            }
        }
    }
}
